package com.it.operator;

import com.it.pojo.Event;
import com.it.operator.utils.SourceUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *  富函数的测试类
 *
 * @author code1997
 */
public class Operator_RichMapFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        eventSource.map(new MyRichFunction()).setParallelism(2).print();
        executionEnvironment.execute();
    }
    public static class MyRichFunction extends RichMapFunction<Event,Integer>{
        /**
         * 当前任务实例被创建的时候调用一次
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("open 生命周期方法被调用 " + getRuntimeContext().getIndexOfThisSubtask());
        }
        /**
         * 当前任务实例被销毁的时候调用一次
         */
        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("close 生命周期方法被调用 " + getRuntimeContext().getIndexOfThisSubtask());
        }

        @Override
        public Integer map(Event value) throws Exception {
            return value.user.length();
        }
    }
}
